package com.tiancy.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;


/***
 * 批处理写法统计单词频次 . 用 DataSet API做了批处理的实现
 * 基本思路：先逐行读入文件数据，然后将每一行文字拆分成单词；接着按照单词分组，统计每组数据的个数，就是对应单词的频次。
 */
public class Using_1_WordCountBatchDemo {
    public static void main(String[] args) throws Exception {

        /*
           TODO 1、创建应用环境
                1.1 批处理环境: 使用 ExecutionEnvironment.getExecutionEnvironment(); 创建批处理环境。这种方法会根据程序的运行环境，返回一个合适的 ExecutionEnvironment。
                如果程序在IDE中运行，它会创建一个 LocalEnvironment，如果程序在集群中运行，它会创建一个 RemoteEnvironment。
                1.2 流处理环境: 使用 StreamExecutionEnvironment.getExecutionEnvironment(); 创建流处理环境。和批处理环境类似，
                这个方法也会根据程序的运行环境，返回一个合适的 StreamExecutionEnvironment。 应用场景：用于流处理任务，对无界数据进行连续的处理。
                1.3 本地环境: 使用 StreamExecutionEnvironment.createLocalEnvironment(); 或者 ExecutionEnvironment.createLocalEnvironment();
                创建本地环境。无论程序在何处运行，这些方法都会创建一个本地环境。应用场景：通常用于开发和测试阶段，因为本地环境可以让你在不连接集群的情况下运行和测试 Flink 程序。
                1.4 除此之外，还可以通过 StreamExecutionEnvironment.createRemoteEnvironment() 或者 ExecutionEnvironment.createRemoteEnvironment() 创建远程环境，
                这需要提供 JobManager 的主机名和端口号，用于连接到指定的 Flink 集群。这种方法通常用于从IDE中提交作业到运行在远程的 Flink 集群。
         */
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // TODO 2、读取文件中的内容,按行读取数据.
        DataSource<String> lineDs = env.readTextFile("input/word.txt");
        // TODO 3、数据切分、转化 (word,1)
        FlatMapOperator<String, Tuple2<String, Integer>> wordToOneDs = lineDs.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    Tuple2<String, Integer> wordToOne = Tuple2.of(word, 1);
                    collector.collect(wordToOne);
                }
            }
        });
        // TODO 3、按照 word 分组 + 聚合
        AggregateOperator<Tuple2<String, Integer>> wordCtn = wordToOneDs.groupBy(0).sum(1);
        // TODO 4、输出打印结果
        wordCtn.print();
    }
}